package cn.itcast.map;


import cn.itcast.bean.SectorBean;
import cn.itcast.config.QuotConfig;
import cn.itcast.constant.Constant;
import cn.itcast.util.DateUtil;
import org.apache.flink.api.common.functions.RichMapFunction;

import java.sql.Timestamp;

public class SectorPutHdfsMap extends RichMapFunction<SectorBean,String> {
    private String sp = QuotConfig.config.getProperty("hdfs.seperator");

    @Override
    public String map(SectorBean sectorBean) throws Exception {
        /**
         * 1.定义字符串字段分隔符
         * 2.日期转换和截取：date类型
         * 3.新建字符串缓存对象
         * 4.封装字符串数据
         */

        //Timestamp|date|sectorCode|sectorName|preClosePrice|openPirce|highPrice|
        //lowPrice|closePrice|tradeVol|tradeAmt|tradeVolDay|tradeAmtDay
        String date = DateUtil.longTimeToString(sectorBean.getEventTime(), Constant.format_yyyy_mm_dd);
        StringBuilder sb = new StringBuilder();
        sb.append(new Timestamp(sectorBean.getEventTime())).append(sp)
                .append(date).append(sp)
                .append(sectorBean.getSectorCode()).append(sp)
                .append(sectorBean.getSectorName()).append(sp)
                .append(sectorBean.getPreClosePrice()).append(sp)
                .append(sectorBean.getOpenPrice()).append(sp)
                .append(sectorBean.getHighPrice()).append(sp)
                .append(sectorBean.getLowPrice()).append(sp)
                .append(sectorBean.getClosePrice()).append(sp)
                .append(sectorBean.getTradeVol()).append(sp)
                .append(sectorBean.getTradeAmt()).append(sp)
                .append(sectorBean.getTradeVolDay()).append(sp)
                .append(sectorBean.getTradeAmtDay());
        return sb.toString();
    }
}
